package eventbus

import (
	"fmt"
	"sync"
)

// 事件
type Event struct {
	Topic   string  // 主题
	Message Message // 消息
}

type Message struct {
	UUID     string
	Metadata Metadata // 元数据
	Payload  Payload  // 负载
}

type Payload []byte

type EventHandler func(event Event)                                  // 消息处理函数
type RequestHandler func(event Event) (response *Message, err error) // 请求处理函数, closeChan 单向通道

// 消息总线定义
type EventBus interface {
	Publish(topic string, message Message)        // 发布信息
	Subscribe(topic string, handler EventHandler) // 订阅消息

	RegisteHandler(topic string, handler RequestHandler) (err error)             // 注册请求处理器
	RequestHandler(topic string, message Message) (response *Message, err error) // 请求应答
}

// 消息总线实现
type EventBusImpl struct {
	handlers map[string][]EventHandler // 消息处理器
	lock     sync.RWMutex              // 读写锁

	// 请求应答
	requestHandlers     map[string]RequestHandler
	requestHandlersLock sync.RWMutex
}

// 创建实例
func NewEventBus() EventBus {
	return &EventBusImpl{
		handlers:        make(map[string][]EventHandler),
		requestHandlers: make(map[string]RequestHandler),
	}
}

// 发布
func (s *EventBusImpl) Publish(topic string, message Message) {
	// 使用协程，不阻塞主线程
	go func() {
		s.lock.RLock()
		defer s.lock.RUnlock()
		for _, handler := range s.handlers[topic] {
			// 调用消息处理程序
			go handler(Event{
				Topic:   topic,
				Message: message,
			})
		}
	}()
}

// 订阅
func (s *EventBusImpl) Subscribe(topic string, handler EventHandler) {
	s.lock.Lock() // 因为要写入，所以锁定
	defer s.lock.Unlock()
	s.handlers[topic] = append(s.handlers[topic], handler)
}

// 注册请求处理器
func (s *EventBusImpl) RegisteHandler(topic string, handler RequestHandler) (err error) {
	s.requestHandlersLock.Lock()
	defer s.requestHandlersLock.Unlock()
	_, ok := s.requestHandlers[topic]
	if ok {
		err = fmt.Errorf("已经存在请求处理器，不能重复注册")
		return
	}
	s.requestHandlers[topic] = handler
	return
}

// 请求应答
func (s *EventBusImpl) RequestHandler(topic string, message Message) (response *Message, err error) { // 请求应答
	s.requestHandlersLock.RLock()
	defer s.requestHandlersLock.RUnlock()
	handler, ok := s.requestHandlers[topic]
	if !ok {
		err = fmt.Errorf("未指定请求处理器")
		return
	}
	//	closeChan := make(chan struct{}, 1)
	response, err = handler(Event{
		Topic:   topic,
		Message: message,
	})
	//	<-closeChan // 等待执行完毕消息

	return
}
